package cn.cihon.ubi.qorosauto.storm;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Map;

import cn.cihon.ubi.qorosauto.biz.VehicleStandardDataFactory;
import cn.cihon.ubi.qorosauto.bo.VehicleStandardData;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class ResolveBolt extends BaseRichBolt{

	/**
	 *   本类用于把Spout传过来的二进制数据解析成明文数据
	 *   并且封装成对象传给下一个Bolt
	 */
	private static final long serialVersionUID = 1L;
	public OutputCollector collerctor;
	public DataInputStream dis = null;
	boolean flag = false;
	public int novinCnt = 0;
	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		this.collerctor = collector;
		
	}

	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		byte[] buffer = new byte[1024];
		 buffer = input.getBinary(0);
		dis = byteToDIS(buffer);
		/*
		 * 开始用的先把文件里的内容读到内存list中，但是发现提交的整个拓扑程序，是先运行完所有的
		 * 读文件到list中（这个过程非常慢），都读到内存中正真的storm集群才开始启动
		 */
		try {
			byte[] itemBuf = new byte[200];
			// 开始对数据进行解析
			while (dis != null) {
				dis.read(itemBuf, 0, 4); // read方法读取一定长度之后，被读取的数据就从流中去掉了，所以下次读取仍然从
											// 0开始
				HeaderFlag = getInt(itemBuf, 0);

				dis.read(itemBuf, 0, 2);
				Version = getShort(itemBuf, 0);

				dis.read(itemBuf, 0, 2);
				RecordLen = getShort(itemBuf, 0);

				dis.read(itemBuf, 0, 17);
				VIN = new String(itemBuf, 0, 17);

				dis.read(itemBuf, 0, 8);
				long a = getLong(itemBuf, 0);
//				RecordTimestamp = getTime(a);

				dis.read(itemBuf, 0, 10);
				UserAccount = new String(itemBuf, 0, 10);

				dis.read(itemBuf, 0, 8);
				long b = getLong(itemBuf, 0);
				CarDataTimestamp = getTime(b);

				dis.read(itemBuf, 0, 4);
				Longitude = getInt(itemBuf, 0) / 3600000.0;

				dis.read(itemBuf, 0, 4);
				Latitude = getInt(itemBuf, 0) / 3600000.0;

				dis.read(itemBuf, 0, 4);
				CanDataLen = getInt(itemBuf, 0);

				dis.read(itemBuf, 0, 4);
				DataPackageSize = getInt(itemBuf, 0);

				dis.read(itemBuf, 0, 8);
				KilometreMileage = getLong(itemBuf, 0);

				dis.read(itemBuf, 0, 2);
				FLPressure = getShort(itemBuf, 0) * 1.373;

				dis.read(itemBuf, 0, 2);
				FRPressure = getShort(itemBuf, 0) * 1.373;

				dis.read(itemBuf, 0, 2);
				RLPressure = getShort(itemBuf, 0) * 1.373;

				dis.read(itemBuf, 0, 2);
				RRPressure = getShort(itemBuf, 0) * 1.373;

				dis.read(itemBuf, 0, 2);
				StatDoorAjarFL = getShort(itemBuf, 0);

				dis.read(itemBuf, 0, 2);
				StatDoorAjarFR = getShort(itemBuf, 0);

				dis.read(itemBuf, 0, 2);
				StatDoorAjarRL = getShort(itemBuf, 0);

				dis.read(itemBuf, 0, 2);
				StatDoorAjarRR = getShort(itemBuf, 0);

				dis.read(itemBuf, 0, 2);
				StatTrunkAjar = getShort(itemBuf, 0);

				dis.read(itemBuf, 0, 4);
				CarDataCount = getInt(itemBuf, 0);

				for (int i = 0; i <= 14; i++) {
					dis.read(itemBuf, 0, 4);
					CarDataSize = getInt(itemBuf, 0);

					dis.read(itemBuf, 0, 4);
					ItemFlags = getInt(itemBuf, 0);

					dis.read(itemBuf, 0, 4);
					int c = getInt(itemBuf, 0);
					CarTime = getTime_1(c);

					dis.read(itemBuf, 0, 8);
					InstantConsum = getLong(itemBuf, 0) * 0.1;

					dis.read(itemBuf, 0, 8);
					LongAcceleration = -(getLong(itemBuf, 0) * 0.02 - 40.96);

					dis.read(itemBuf, 0, 2);
					SteeringPos = getShort(itemBuf, 0) * 0.1 - 800;

					dis.read(itemBuf, 0, 4);
					SteeringAnglleSpeed = getInt(itemBuf, 0) * 0.044 - 1440;

					dis.read(itemBuf, 0, 2);
					SteeringPosState = getShort(itemBuf, 0) * 1.0;

					dis.read(itemBuf, 0, 2);
					STATSASCalibration = getShort(itemBuf, 0) * 1.0;

					dis.read(itemBuf, 0, 8);
					VehicleSpeed = getLong(itemBuf, 0) * 0.075;

					dis.read(itemBuf, 0, 8);
					EngineSpeed = getLong(itemBuf, 0) * 0.25;
					
					/*
					 * 已知每秒必会获取15条数据，为了正确地表示15秒内每一条数据的时间，
					 * 在recordTimestamp上加循环变量 i * 1000 毫秒
					 */
					RecordTimestamp = getTime(a,i);

					String str = HeaderFlag + "," + Version + "," + RecordLen
							+ "," + VIN + "," + RecordTimestamp + ","
							+ UserAccount + "," + CarDataTimestamp + ","
							+ Longitude + "," + Latitude + "," + CanDataLen
							+ "," + DataPackageSize + "," + KilometreMileage
							+ "," + FLPressure + "," + FRPressure + ","
							+ RLPressure + "," + RRPressure + ","
							+ StatDoorAjarFL + "," + StatDoorAjarFR + ","
							+ StatDoorAjarRL + "," + StatDoorAjarRR + ","
							+ StatTrunkAjar + "," + CarDataCount + ","
							+ CarDataSize + "," + ItemFlags + "," + CarTime
							+ "," + InstantConsum + "," + LongAcceleration
							+ "," + SteeringPos + "," + SteeringAnglleSpeed
							+ "," + SteeringPosState + "," + STATSASCalibration
							+ "," + VehicleSpeed + "," + EngineSpeed;
					
					if (VIN == null || "".equals(VIN.trim())) {
						novinCnt++;
						throw new Exception("发现空VIN码数据，目前出现了" + novinCnt
								+ "条!datastr：" + str + "!");
					}
					String[] strs = str.split(",");
					// 将tuple中的非标准明文数据封装到标准对象
					VehicleStandardData stdData = VehicleStandardDataFactory
							.createVehicleStandardData(strs);
					ArrayList list = new ArrayList();
					list.add(VIN);
					list.add(stdData);
					this.collerctor.emit(input, new Values((String) list.get(0),
							(VehicleStandardData) list.get(1)));

					// this.outputCollector.emit(new Values(str),str);
					// //发送已经解析好的数据
				}
				dis.read(itemBuf, 0, 117);
				String s = new String(itemBuf, 0, 117);
				break;
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		
		this.collerctor.ack(input);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("VIN", "stdData")); // 声明streamId
	}
	
	public int HeaderFlag = 0;
	public short Version = 0;
	public short RecordLen = 0;
	public String VIN = null;
	public String RecordTimestamp = null;
	public String UserAccount = null;
	public String CarDataTimestamp = null;
	public double Longitude = 0;
	public double Latitude = 0;
	public int CanDataLen = 0;
	public int DataPackageSize = 0;
	public long KilometreMileage = 0;
	public double FLPressure = 0;
	public double FRPressure = 0;
	public double RLPressure = 0;
	public double RRPressure = 0;
	public short StatDoorAjarFL = 0;
	public short StatDoorAjarFR = 0;
	public short StatDoorAjarRL = 0;
	public short StatDoorAjarRR = 0;
	public short StatTrunkAjar = 0;
	public int CarDataCount = 0;
	public int CarDataSize = 0;
	public int ItemFlags = 0;
	public String CarTime = null;
	public double InstantConsum = 0;
	public double LongAcceleration = 0;
	public double SteeringPos = 0;
	public double SteeringAnglleSpeed = 0;
	public double SteeringPosState = 0;
	public double STATSASCalibration = 0;
	public double VehicleSpeed = 0;
	public double EngineSpeed = 0;

	// byte数组转换成short类型
	public static short getShort(byte[] b, int index) {
		return (short) (((b[index + 1] << 8) | b[index + 0] & 0xff));
	}

	// byte数组转换成int类型
	public static int getInt(byte[] bb, int index) {
		return (int) ((((bb[index + 3] & 0xff) << 24)
				| ((bb[index + 2] & 0xff) << 16)
				| ((bb[index + 1] & 0xff) << 8) | ((bb[index + 0] & 0xff) << 0)));
	}

	// byte数组转换成long类型
	public static long getLong(byte[] bb, int index) {
		return ((((long) bb[index + 7] & 0xff) << 56)
				| (((long) bb[index + 6] & 0xff) << 48)
				| (((long) bb[index + 5] & 0xff) << 40)
				| (((long) bb[index + 4] & 0xff) << 32)
				| (((long) bb[index + 3] & 0xff) << 24)
				| (((long) bb[index + 2] & 0xff) << 16)
				| (((long) bb[index + 1] & 0xff) << 8) | (((long) bb[index + 0] & 0xff) << 0));
	}

	// long类型转换成时间
		public static String getTime(long num) {
			long dd = 621355968000000000L;
			num = (num - dd) / 10000 + 8 * 3600000; // c#与java时间转换
			Date mydate = new Date(num);
			SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			String etime = format.format(mydate);
			return etime;
		}
		
	// long类型转换成时间
	public static String getTime(long num,int i) {
		long dd = 621355968000000000L;
		num = (num - dd) / 10000 + 8 * 3600000+i*1000; // c#与java时间转换
		Date mydate = new Date(num);
		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		String etime = format.format(mydate);
		return etime;
	}

	public static String getTime_1(int num) {
		long tim = (long) (num + 946656000) * 1000; // java时间转换和unix时间戳
		Date mydate = new Date(tim);
		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		String etime = format.format(mydate);
		return etime;
	}

	public static String getcarstr(String str) {
		return str;
	}
	
	public DataInputStream byteToDIS(byte[] b) {
		DataInputStream dis = new DataInputStream(new ByteArrayInputStream(b));
		return dis;
	}
}
